package com.shujia.spark.opt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo2RePartition {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setMaster("local[8]")
      .setAppName("partition")

    val sc = new SparkContext(conf)

    /**
      * 读取hdfs文件得到的rdd分区数由block决定
      */

    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    println(s"studentRDD分区数：${studentRDD.getNumPartitions}")

    /**
      * 在使用shuffle类算子的时候可以改变分区数
      *
      */

    /**
      * repartition: 修改rdd的分区数，没有具体代码逻辑
      * repartition会产生shuffle
      *
      */

    //coalesce(numPartitions, shuffle = true)
    val rePartitionRDD: RDD[String] = studentRDD.repartition(10)


    println(s"rePartitionRDD分区数：${rePartitionRDD.getNumPartitions}")

    /**
      * coalesce: 指定shuffle为false 只能用于减少分区  -- 一般用于合并小文件
      * coalesce： 指定shuffle为true 可以用于增加分区和减少分区 -- 提高并行度
      */


    val coalesceRDD: RDD[String] = rePartitionRDD.coalesce(20, false)

    println(s"coalesceRDD分区数：${coalesceRDD.getNumPartitions}")


    //产生小文件
    val rdd1: RDD[String] = studentRDD.repartition(200)

    rdd1.saveAsTextFile("data/partition")

  }

}
